#include "execution/executors/sort_executor.h"
#include <algorithm>
#include <utility>
#include "binder/bound_order_by.h"
#include "common/macros.h"
#include "common/rid.h"
#include "execution/plans/abstract_plan.h"
#include "storage/table/tuple.h"
#include "type/type.h"

namespace bustub {

SortExecutor::SortExecutor(ExecutorContext *exec_ctx, const SortPlanNode *plan,
                           std::unique_ptr<AbstractExecutor> &&child_executor)
    : AbstractExecutor(exec_ctx), plan_(plan), child_executor_(std::move(child_executor)) {}

void SortExecutor::Init() {
  child_executor_->Init();
  Tuple tuple{};
  RID dummy_rid{};
  while (child_executor_->Next(&tuple, &dummy_rid)) {
    tuples_.push_back(std::move(tuple));
  }

  std::sort(tuples_.begin(), tuples_.end(), [this](const Tuple &a, const Tuple &b) {
    for (auto &order_by : plan_->GetOrderBy()) {
      auto a_key = order_by.second->Evaluate(&a, GetOutputSchema());
      auto b_key = order_by.second->Evaluate(&b, GetOutputSchema());
      if (order_by.first == OrderByType::ASC || order_by.first == OrderByType::DEFAULT) {
        if (a_key.CompareLessThan(b_key) == CmpBool::CmpTrue) {
          return true;
        }
        if (a_key.CompareGreaterThan(b_key) == CmpBool::CmpTrue) {
          return false;
        }
      } else if (order_by.first == OrderByType::DESC) {
        if (a_key.CompareGreaterThan(b_key) == CmpBool::CmpTrue) {
          return true;
        }
        if (a_key.CompareLessThan(b_key) == CmpBool::CmpTrue) {
          return false;
        }
      }
    }
    UNREACHABLE("doesn't support duplicate key");
  });

  tuples_iterator_ = tuples_.begin();
}

auto SortExecutor::Next(Tuple *tuple, RID *rid) -> bool {
  if (tuples_iterator_ == tuples_.end()) {
    return false;
  }
  *tuple = *tuples_iterator_;
  ++tuples_iterator_;
  return true;
}

}  // namespace bustub
